package service

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/google/uuid"
	"github.com/pkg/errors"
	"gorm.io/gorm"
	"inventory/api/inventory"
	"inventory/api/mall"
	"inventory/api/prod"
	"inventory/internal/conf"
	"inventory/internal/constant"
	"inventory/pkg/nsq"
	"inventory/pkg/request_key"
	"inventory/pkg/util"
	"inventory/pkg/wechat"
	"time"

	"inventory/internal/model"
	"inventory/pkg/log"
)

type msgRetryKey struct{}

func (s *Service) saveOrUpdateError(ctx context.Context, logBody string, msgBody []byte, rawError *error) {
	now := time.Now()
	retryMessage, _ := ctx.Value(msgRetryKey{}).(*model.NsqMessage)
	topic, _ := ctx.Value(nsq.TopicContextKey).(string)
	if *rawError != nil {
		// 只要失败，打印日志，仅保留单号记录
		log.FromContext(ctx).Warnw("nsq",
			"status_code", "Fail-Unknown",
			"method", "consume",
			"path", topic,
			"body", logBody,
			"error", (*rawError).Error(),
			"stacktrace", fmt.Sprintf("postgres error: %+v", *rawError),
		)
		if retryMessage == nil {
			executeTime := now.Add(time.Minute)
			if err := s.dao.Create(ctx, &model.NsqMessage{
				Topic:       topic,
				MsgKey:      uuid.New().String(), // 这里防止订单更新消息落地失败，失败消息key置换为uuid
				Payload:     msgBody,
				ExecuteTime: &executeTime,
				CreateTime:  now,
				UpdateTime:  now,
			}); err != nil {
				log.FromContext(ctx).Infow("nsq",
					"status_code", "Fail-Unknown",
					"method", "errorSaving",
					"path", topic,
					"body", logBody,
					"error", err.Error(),
					"stacktrace", fmt.Sprintf("postgres error: %+v", err),
				)
			}
			// 终止外部报错，允许消息正常结束；依赖定时任务触发后续步骤，不触发企业微信
			*rawError = nil
		} else {
			executeTime := now.Add(time.Minute * time.Duration(retryMessage.Retry*retryMessage.Retry))
			retryMessage.Retry += 1
			retryMessage.ExecuteTime = &executeTime
			retryMessage.UpdateTime = now
			if err := s.dao.Save(ctx, retryMessage); err != nil {
				log.FromContext(ctx).Infow("nsq",
					"status_code", "Fail-Unknown",
					"method", "errorUpdating",
					"path", topic,
					"body", logBody,
					"error", err.Error(),
					"stacktrace", fmt.Sprintf("postgres error: %+v", err),
				)
			}
			if retryMessage.Retry > 30 {
				// 重试超过30次，触发企业微信报警
				wechat.AsyncSendMarkDown(
					ctx, wechat.FormatBugMessage(
						ctx, conf.Conf.Env, conf.Conf.ServiceName, topic,
						string(msgBody),
						fmt.Sprintf("postgres error: %+v", *rawError)))
			}
		}
	} else {
		// 成功时打印简短日志，保留单号记录
		log.FromContext(ctx).Infow("nsq",
			"status_code", "Success",
			"method", "consume",
			"path", topic,
			"body", logBody,
		)
		if retryMessage != nil {
			// 已发送成功，更新重试任务
			retryMessage.ExecuteTime = nil
			retryMessage.UpdateTime = now
			retryMessage.Status = 1
			if err := s.dao.Save(ctx, retryMessage); err != nil {
				log.FromContext(ctx).Infow("nsq",
					"status_code", "Fail-Unknown",
					"method", "errorUpdating",
					"path", topic,
					"body", logBody,
					"error", err.Error(),
					"stacktrace", fmt.Sprintf("postgres error: %+v", err),
				)
			}
		}
	}
	return
}

// 批量更新商品信息
func (s *Service) BatchUpdateProdsHandler(ctx context.Context, msgBody []byte) (err error) {
	req := &prod.ProdUpdateMessage{}
	if err = json.Unmarshal(msgBody, req); err != nil {
		return err
	}
	defer s.saveOrUpdateError(ctx, fmt.Sprintf("prod_update: %s", string(msgBody)),
		msgBody, &err)
	prodIds := util.GetUniqueSlice(req.Ids)
	var prods []*prod.SelectionProdView
	if prods, err = s.downstream.FetchCommonProds(ctx, req.SgId, prodIds); err != nil {
		return err
	}
	err = s.dao.Transaction(ctx, func(ctx context.Context) (err error) {
		now := time.Now()
		for _, p := range prods {
			skuStorageKeys := make([]string, 0)
			for _, sku := range p.Skus {
				storageKey := fmt.Sprintf("%d-%d", p.Id, sku.Id)
				baseRemark := model.Remark{
					Code:    storageKey,
					Unit:    p.Unit,
					Name:    p.Name,
					SkuName: sku.Name,
				}
				skuStorageKeys = append(skuStorageKeys, storageKey)
				remark := &model.StorageRealTimeRemark{
					Remark:        baseRemark,
					Type:          p.ProdTypeId,
					StorageUnit:   p.Unit,
					StorageRatio:  1,
					PurchaseUnit:  p.Unit,
					PurchaseRatio: 1,
				}
				remarkJSON, _ := json.Marshal(remark)
				if err = s.dao.Update(ctx, model.Storage{StorageKey: storageKey}, map[string]interface{}{
					"remark": remarkJSON,
				}); err != nil {
					return
				}
			}
			// 清理已删除规格库存项
			if err = s.dao.TransContext(ctx).Delete(&model.Storage{},
				"prod_id = ? and storage_key not in ?",
				p.Id, skuStorageKeys).Error; err != nil {
				return
			}
			// 删除bom配置
			if err = s.dao.TransContext(ctx).Model(model.BomConfig{}).Where(
				"sg_id = ? and ref_id = ? and storage_type = ? and storage_key not in ? and status = ?",
				req.SgId, p.Id, inventory.StorageType_PROD, skuStorageKeys,
				constant.BomValid).Updates(map[string]interface{}{
				"status":      constant.BomDeleted,
				"update_time": now,
			}).Error; err != nil {
				return
			}
		}
		return
	})
	return
}

// 批量删除产品
func (s *Service) BatchDeleteProdsHandler(ctx context.Context, msgBody []byte) (err error) {
	// 批量删除商品信息
	req := &prod.ProdDeleteMessage{}
	if err = json.Unmarshal(msgBody, req); err != nil {
		return err
	}
	defer s.saveOrUpdateError(ctx, fmt.Sprintf("prod_delete: %s", string(msgBody)),
		msgBody, &err)
	prodIDs := util.GetUniqueSlice(req.Ids)
	err = s.dao.Transaction(ctx, func(ctx context.Context) (err error) {
		if err = s.dao.TransContext(ctx).Delete(&model.Storage{}, "prod_id in ?",
			prodIDs).Error; err != nil {
			return
		}
		if err = s.dao.TransContext(ctx).Model(&model.BomConfig{}).Where("sg_id = ? and status = 1 and ref_id in ?",
			req.SgId, prodIDs).Updates(map[string]interface{}{
			"status":      0,
			"update_time": time.Unix(int64(req.Timestamp), 0),
		}).Error; err != nil {
			return
		}
		return
	})
	return
}

// 履约单被动更新（商城、星团确认收货/退款标记/退款标记取消）
func (s *Service) OrderUpdateHandler(ctx context.Context, msgBody []byte) (err error) {
	req := &inventory.InventoryOrderNotifyPayload{}
	if err = json.Unmarshal(msgBody, req); err != nil {
		return err
	}
	defer s.saveOrUpdateError(ctx, fmt.Sprintf("order_update: %s", req.OrderNumber),
		msgBody, &err)
	err = s.dao.Transaction(ctx, func(ctx context.Context) (err error) {
		if req.OrderNumber == "" {
			err = fmt.Errorf("单号不能为空")
			return
		}
		var processOrder *model.ProcessOrder
		if processOrder, err = s.dao.LockProcessOrderByExtOrderNumberAndSource(
			ctx, req.OrderNumber, int(req.Source)); err != nil {
			if errors.Cause(err) == gorm.ErrRecordNotFound {
				err = constant.ErrOrderNotExist
			}
			return
		}
		now := time.Unix(req.Timestamp, 0)
		if processOrder.UpdateTime.Unix() >= req.Timestamp {
			log.FromContext(ctx).Warn(fmt.Sprintf("NSQ msg outdated, input-%s, db-%s",
				now.String(), processOrder.UpdateTime.String()))
			return
		}
		switch req.Action {
		case "received":
			// 被动接受消息时，跳过前置检查
			switch inventory.ProcessOrderStatus(processOrder.Status) {
			case inventory.ProcessOrderStatus_ProcessOrderWaitMentioned:
				processOrder.Status = uint8(inventory.ProcessOrderStatus_ProcessOrderReceived)
				_ = processOrder.ExtractKey.Scan(nil)
				var warehouse *model.Warehouse
				if warehouse, err = s.dao.GetWarehouseInfoInternal(ctx, processOrder.WarehouseID); err != nil {
					return
				}
				deadlineTime, _ := warehouse.Deadline.Value()
				var prepareTime string
				if processOrder.PrepareTime.Valid {
					prepareTime = processOrder.PrepareTime.Time.Format(util.DefaultLayout)
				}
				if err = deliverProcessStorage(ctx, s.dao, processOrder, &inventory.InventoryOrderUpdatePayload_DeliverMemo{
					ExtractKey:    fmt.Sprint(processOrder.ExtractKey.String),
					DeadlineTime:  fmt.Sprintf("%s %s", now.Format(util.DateLayout), deadlineTime),
					PickUpAddress: warehouse.Address,
					DeliverTime:   now.Format(util.DefaultLayout),
					PrepareTime:   prepareTime,
				}, now, false); err != nil {
					return
				}
				return
			case inventory.ProcessOrderStatus_ProcessOrderShipped:
				// 终态不允许标记
				if err = s.dao.Update(ctx, &processOrder, map[string]interface{}{
					"update_time": now,
					"status":      int(inventory.ProcessOrderStatus_ProcessOrderReceived),
				}); err != nil {
					return
				}
				deliverMemo := &inventory.InventoryOrderUpdatePayload_DeliverMemo{
					DeliverTime: processOrder.DeliverTime.Time.Format(util.DefaultLayout),
				}
				var (
					expressOrder *model.ExpressOrder
					platform     *model.ExpressPlatform
				)
				if expressOrder, err = s.dao.GetExpressOrderByID(ctx, uint64(processOrder.ExpressOrderID.Int64)); err != nil {
					return
				}
				if err = s.dao.TransContext(ctx).Model(&model.ExpressOrder{}).Where(
					"id = ?", int(processOrder.ExpressOrderID.Int64)).Updates(
					map[string]interface{}{
						"status":      int(inventory.DeliverStatus_ExpressOrderReceived),
						"update_time": now,
					}).Error; err != nil {
					return
				}
				if platform, err = s.dao.GetExpressPlatform(ctx, expressOrder.PlatformID); err != nil {
					return
				}
				var prepareTime string
				if processOrder.PrepareTime.Valid {
					prepareTime = processOrder.PrepareTime.Time.Format(util.DefaultLayout)
				}
				deliverMemo.PlatformOrderNum = expressOrder.TradeNo
				deliverMemo.Platform = platform.Name
				deliverMemo.PrepareTime = prepareTime
				statsMsg, _ := json.Marshal(&inventory.InventoryOrderUpdatePayload{
					OrderNumber:    processOrder.OrderNumber,
					ExtOrderNumber: processOrder.ExtOrderNumber,
					Source:         inventory.SourceEnum(processOrder.Source),
					Timestamp:      now.Unix(),
					Status:         inventory.ProcessOrderStatus(processOrder.Status),
					DeliverType:    inventory.DeliverType(processOrder.DeliverType),
					DeliverMemo:    deliverMemo,
				})
				if err = s.dao.PublishUniqueNSQMessage(ctx, constant.TopicInventoryStatus, statsMsg); err != nil {
					return err
				}
				return
			default:
				// 其他状态成功，丢弃消息不做标记
				log.FromContext(ctx).Warn(fmt.Sprintf("NSQ Input Duplicated: raw_status=%d, action=%s",
					processOrder.Status, req.Action))
				return
			}
		case "refund_apply", "refund_cancel":
			switch inventory.ProcessOrderStatus(processOrder.Status) {
			case inventory.ProcessOrderStatus_ProcessOrderCancelled, inventory.ProcessOrderStatus_ProcessOrderReturned,
				inventory.ProcessOrderStatus_ProcessOrderRejectReturn:
				// 终态幂等成功，不做更新
				log.FromContext(ctx).Warn(fmt.Sprintf("NSQ Input Duplicated: raw_status=%d, action=%s",
					processOrder.Status, req.Action))
				return
			default:
				var refundFlag int
				if req.Action == "refund_apply" {
					refundFlag = 1
				}
				if err = s.dao.Update(ctx, &processOrder, map[string]interface{}{
					"update_time": now,
					"refund_flag": refundFlag,
				}); err != nil {
					return
				}
				return
			}
		default:
			err = fmt.Errorf("订单操作未配置")
			return
		}
	})
	return
}

//// 门店更新
//func (s *Service) StoreUpdateHandler(ctx context.Context, msgBody []byte) (err error) {
//	// 门店同步
//	req := &bilinadmin.StoreUpdateMessage{}
//	if err = json.Unmarshal(msgBody, req); err != nil {
//		return err
//	}
//	defer s.saveOrUpdateError(ctx, fmt.Sprintf("store_update: %d", req.StoreId),
//		msgBody, &err)
//	err = s.dao.Transaction(ctx, func(ctx context.Context) (err error) {
//		var warehouse *model.Warehouse
//		if warehouse, err = s.dao.GetWarehouseInfoByStoreID(ctx, uint32(req.StoreId)); err != nil {
//			if errors.Cause(err) == gorm.ErrRecordNotFound {
//				// 未建立仓库，无需更新
//				err = nil
//				return
//			}
//			return
//		}
//		if warehouse.SyncTime != nil && warehouse.SyncTime.Unix() >= req.Timestamp {
//			// 丢弃历史消息
//			log.FromContext(ctx).Warn(fmt.Sprintf("NSQ msg outdated, input-%s, db-%s",
//				time.Unix(int64(req.Timestamp), 0).String(), warehouse.SyncTime.String()))
//			return
//		}
//		warehouse.Name = req.StoreName
//		warehouse.Address = req.Address
//		warehouse.Lng = req.Longitude
//		warehouse.Lat = req.Latitude
//		cityCode, _ := strconv.Atoi(req.DistrictCode)
//		warehouse.CityCode = uint32(cityCode)
//		warehouse.RegionCode = req.RegionCode
//		warehouse.Contact = req.ContactName
//		warehouse.Mobile = req.ContactMobile
//		if warehouse.Status == constant.WarehouseAbnormal {
//			if req.Latitude != 0 && req.Longitude != 0 && warehouse.CityCode != 0 && req.RegionCode != "" && req.Address != "" {
//				warehouse.Status = constant.WarehouseEnable
//			}
//		}
//		warehouse.Operator = uint32(req.EmpId)
//		syncTime := time.Unix(req.Timestamp, 0)
//		warehouse.SyncTime = &syncTime
//		if err = s.dao.Save(ctx, warehouse); err != nil {
//			return
//		}
//		return
//	})
//	return
//}

// 门店更新
func (s *Service) MallBizCancelHandler(ctx context.Context, msgBody []byte) (err error) {
	// 门店同步
	req := &mall.MallBizCancelMessage{}
	if err = json.Unmarshal(msgBody, req); err != nil {
		return err
	}
	defer s.saveOrUpdateError(ctx, fmt.Sprintf("mall_biz_cancel: %s", req.BizNo),
		msgBody, &err)
	if _, err = s.EndFreezeOrder(ctx, &inventory.EndFreezeOrderRequest{
		RequestNum: req.BizNo,
		Source:     inventory.SourceEnum_MALL_ORDER,
	}); err != nil {
		return
	}
	return
}

// 分摊结果
func (s *Service) PromotionResultHandler(ctx context.Context, msgBody []byte) (err error) {
	req := &mall.MallPromotionMessage{}
	if err = json.Unmarshal(msgBody, req); err != nil {
		return err
	}
	defer s.saveOrUpdateError(ctx, fmt.Sprintf("promotion_notify: %s", string(msgBody)),
		msgBody, &err)
	if err = s.dao.Transaction(ctx, func(ctx context.Context) (err error) {
		var details []*model.ProcessDetail
		var order *model.ProcessOrder
		if order, err = s.dao.LockProcessOrderByExtOrderNumberAndSource(
			ctx, req.OrderNumber, int(inventory.SourceEnum_MALL_ORDER)); err != nil {
			if errors.Cause(err) == gorm.ErrRecordNotFound {
				// 存在商城有分摊但无发货单无需更新分摊的场景
				err = nil
			}
			return
		}
		if order.PromotionStatus {
			// 冗余推送
			return
		}
		promotionMap := make(map[string]uint64)
		for _, detail := range req.Details {
			promotionMap[detail.StorageKey] = detail.Off
		}
		if details, err = s.dao.FindProcessOrderDetails(ctx, order.ID); err != nil {
			return
		}
		for _, detail := range details {
			if off, ok := promotionMap[detail.StorageKey]; ok {
				detail.Discount = int64(off)
				detail.Trans = detail.Origin - detail.Discount
				if err = s.dao.Save(ctx, detail); err != nil {
					return
				}
				delete(promotionMap, detail.StorageKey)
			}
		}
		if len(promotionMap) != 0 {
			err = fmt.Errorf("促销推送订单明细与商城订单不匹配")
			return
		}
		return
	}); err != nil {
		return
	}
	return
}

// NSQ消息恢复
func (s *Service) NSQRetry(ctx context.Context, req *inventory.CommonRetryRequest) (resp *inventory.DefaultResp, err error) {
	limit := int(req.Limit)
	if limit == 0 {
		limit = 1000
	}
	var messages, sendMsgs []*model.NsqMessage
	if messages, err = s.dao.FindRetryMessages(ctx, limit); err != nil {
		return
	}
	// 标记上下文为retry版本
	if len(messages) != 0 {
		requestId := request_key.GetRequestID(ctx)
		// 异步执行，不阻塞请求，防止外部请求超时时，context cancel导致后续处理失败
		go func() {
			for _, msg := range messages {
				if _, ok := processorMap[msg.Topic]; ok {
					// 消息接收重试任务
					subCtx := request_key.GenerateContextWithRequestID(requestId)
					subCtx = context.WithValue(subCtx, msgRetryKey{}, msg)
					if msgError := processorMap[msg.Topic](subCtx, msg.Payload); msgError != nil {
						log.FromContext(subCtx).Infow("nsq",
							"status_code", "Fail-Unknown",
							"method", "retry",
							"path", msg.Topic,
							"body", msg.MsgKey,
							"error", msgError.Error(),
							"stacktrace", fmt.Sprintf("msg retry error: %+v", err),
						)
					}
				} else {
					// 消息发送重试任务
					sendMsgs = append(sendMsgs, msg)
				}
			}
			if len(sendMsgs) != 0 {
				// 批量重试
				subCtx := request_key.GenerateContextWithRequestID(requestId)
				err = s.dao.Transaction(subCtx, func(subCtx context.Context) (err error) {
					err = s.dao.RetryNSQMessages(subCtx, sendMsgs)
					return
				})
			}
		}()
	}
	resp = &inventory.DefaultResp{Status: true}
	return
}
